using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api.Exceptions;
using Org.Apache.Cassandra.Thrift;
using Thrift;
using Thrift.Protocol;
using Thrift.Transport;
using Thrift.Server;
using Me.Prettyprint.Cassandra.Utils;

namespace Me.Prettyprint.Cassandra.Connection.Client
{
    /// <summary>
    /// Thrift implementation for hector client.
    /// </summary>
	public class HThriftClient : HClient
	{
        private static long serial = 0L;
		
		private const string NAME_FORMAT = "CassandraClient<{0}-{1}>";
		internal CassandraHost cassandraHost;
		internal ExceptionsTranslator exceptionsTranslator;
		protected internal int timeout;
		protected internal string keyspaceName;
		private long useageStartTime;
        private long mySerial;
		
		protected internal TTransport transport;
		protected internal Org.Apache.Cassandra.Thrift.Cassandra.Client cassandraClient;
        private Logger logger = new Logger(typeof(HThriftClient));

        public HThriftClient(CassandraHost cassandraHost)
        {
            ClearAuthentication();
            CreatedTime = (System.DateTime.Now.Ticks - 621355968000000000) / 10000;
            this.cassandraHost = cassandraHost;
            this.timeout = GetTimeout(cassandraHost);
            mySerial = Interlocked.Increment(ref serial);
            exceptionsTranslator = new ExceptionsTranslatorImpl();
        }
        public void SetAuthentication(Dictionary<string, string> creds)
        {
            Credentials = creds;
        }
        public Dictionary<string, string> Credentials
        {
            get;
            set;
        }
        public void ClearAuthentication()
        {
            if (Credentials != null && Credentials.Count > 0)
            {
                Credentials.Clear();
            }
        }
        public bool AlreadyAuthenticated(Dictionary<string, string> creds)
        {
            return creds != null && Credentials.Equals(creds);
        }
		public bool IsOpen()
		{
            bool open = false;
            if (transport != null)
            {
                open = transport.IsOpen;
            }
            return open;
        }
		public long SinceLastUsed
		{
			get
			{
				return (System.DateTime.Now.Ticks - 621355968000000000) / 10000 - useageStartTime;
			}
		}
		public CassandraHost CassandraHost
		{
			get
			{
				return cassandraHost;
			}
		}
        public long CreatedTime
        {
            get;
            internal set;
        }
        public long LastSuccessTime
        {
            get;
            internal set;
        }
		public Org.Apache.Cassandra.Thrift.Cassandra.Client GetCassandra()
		{
			if (!IsOpen())
			{
				throw new System.SystemException("getCassandra called on client that was not open. You should not have gotten here.");
			}
			if (cassandraClient == null)
			{
				cassandraClient = new Org.Apache.Cassandra.Thrift.Cassandra.Client(new TBinaryProtocol(transport));
			}
			return cassandraClient;
		}
		public Org.Apache.Cassandra.Thrift.Cassandra.Client GetCassandra(string keyspaceNameArg)
		{
			GetCassandra();
			if (keyspaceNameArg != null)
			{
				try
				{
					cassandraClient.set_keyspace(keyspaceNameArg);
				}
				catch (InvalidRequestException ire)
				{
					throw new HInvalidRequestException(ire);
				}
				keyspaceName = keyspaceNameArg;
			}
			return cassandraClient;
		}
        public HClient Close()
		{
			if (IsOpen())
			{
				try
				{
                    transport.Flush();
				}
				catch (System.Exception e)
				{
				}
				finally
				{
					try
					{
                        transport.Close();
					}
					catch (System.Exception e)
					{
                        logger.Warning(e.Message);
					}
				}
			}
			return this;
		}
        public HClient Open()
		{
			if (IsOpen())
			{
				throw new System.SystemException("Open called on already open connection. You should not have gotten here.");
			}
			
			TSocket socket;
			try
			{
				socket = new TSocket(cassandraHost.Host, cassandraHost.Port, timeout);
			}
			catch (TTransportException e)
			{
				throw new HectorTransportException("Could not get client socket: ", e);
			}
			
			if (cassandraHost.UseSocketKeepalive)
			{
				try
				{
                    socket.TcpClient.Client.SetSocketOption(System.Net.Sockets.SocketOptionLevel.Tcp, System.Net.Sockets.SocketOptionName.KeepAlive, true);
				}
				catch (System.Net.Sockets.SocketException se)
				{
					throw new HectorTransportException("Could not set socket options for KEEPALIVE on socket: ", se);
				}
			}
			
			transport = MaybeWrapWithTFramedTransport(socket);
			
			// If using SSL, the socket will already be connected, and TFramedTransport and
			// TSocket just wind up calling socket.isConnected(), so check this before calling
			// open() to avoid a "Socket already connected" error.
			if (!transport.IsOpen)
			{
				try
				{
                    transport.Open();
				}
				catch (TTransportException e)
				{
					// Thrift exceptions aren't very good in reporting, so we have to catch the exception here and
					// add details to it.
					throw new HectorTransportException("Unable to open transport to " + cassandraHost.Name + " , " + e.Message, e);
				}
			}
			return this;
		}
		protected internal TTransport MaybeWrapWithTFramedTransport(TTransport transport)
		{
			if (cassandraHost.UseThriftFramedTransport)
			{
				return new TFramedTransport(transport);
			}
			else
			{
				return transport;
			}
		}
		private int GetTimeout(CassandraHost cassandraHost)
		{
			int timeoutVar = 0;
			if (cassandraHost != null && cassandraHost.CassandraThriftSocketTimeout > 0)
			{
				timeoutVar = cassandraHost.CassandraThriftSocketTimeout;
			}
			return timeoutVar;
		}
		public void StartToUse()
		{
			useageStartTime = (System.DateTime.Now.Ticks - 621355968000000000) / 10000;
		}
		public override string ToString()
		{
			return string.Format(NAME_FORMAT, cassandraHost.Url, mySerial);
		}
		public override bool Equals(System.Object obj)
		{
			return this.ToString().Equals(obj.ToString());
		}
		public void UpdateLastSuccessTime()
		{
			LastSuccessTime = (System.DateTime.Now.Ticks - 621355968000000000) / 10000;
		}
        public override int GetHashCode()
		{
			return base.GetHashCode();
		}
	}
}